-module(emqx_plugin_kafka).

-include_lib("emqx/include/logger.hrl").
-include_lib("emqx/include/emqx_hooks.hrl").
-include_lib("emqx/include/emqx.hrl").

-export([load/1, unload/0]).
-export([on_client_connected/3, on_client_disconnected/4]).
-export([on_message_publish/2]).
% #{<<"plugin_kafka">> => #{<<"bootstrap_hosts">> => [<<"10.1.86.200:30092">>],
% <<"mqtt_topics">> => [<<"test/#">>],<<"query_api_versions">> => true,
% <<"reconnect_cool_down_seconds">> => 15,<<"topic">> => <<"test">>}} ! mfa: undefined
load(Cnf) ->
  % #{bootstrap_hosts => [<<"10.1.86.200:30092">>],mqtt_topics => [<<"test/#">>],query_api_versions => true,reconnect_cool_down_seconds => 15,topic => <<"test">>}
    kafka_init(Cnf),
    % 挂载钩子
    % hook('client.connect',  {?MODULE, on_client_connect, [Cnf]}),  % 服务端收到客户端连接报文
    % hook('client.connack',      {?MODULE, on_client_connack, [Cnf]}),  % 下发连接应答
    hook('client.connected',    {?MODULE, on_client_connected, [Cnf]}), % 完成认证并成功接入系统
    hook('client.disconnected', {?MODULE, on_client_disconnected, [Cnf]}), % 客户端连接层准备关闭时
    % hook('session.subscribed',  {?MODULE, on_session_subscribed, [Cnf]}), % 收到订阅报文后，执行 client.authorize 鉴权前
    % hook('session.unsubscribed',{?MODULE, on_session_unsubscribed, [Cnf]}), % 收到取消订阅报文后
    hook('message.publish',     {?MODULE, on_message_publish, [Cnf]}).  % 服务端在发布（路由）消息前
    % hook('message.delivered',     {?MODULE, on_message_delivered, [Cnf]}),  % 消息准备投递到客户端前
    % hook('message.acked',       {?MODULE, on_message_acked, [Cnf]}), % 服务端在收到客户端发回的消息 ACK 后  
    % hook('message.dropped',     {?MODULE, on_message_dropped, [Cnf]}). % 	发布出的消息被丢弃后



hook(HookPoint, Action) ->
    emqx_hooks:add(HookPoint, Action, _Property = ?HP_HIGHEST).

unload() -> 
    %  卸载钩子
    unhook('client.connected',    {?MODULE, on_client_connected}),
    unhook('client.disconnected', {?MODULE, on_client_disconnected}),
    % unhook('session.subscribed',{?MODULE, on_session_subscribed}),
    % unhook('session.unsubscribed',{?MODULE, on_session_unsubscribed}),
    unhook('message.publish',     {?MODULE, on_message_publish}),
    % unhook('message.delivered',     {?MODULE, on_message_delivered}),
    % unhook('message.acked',       {?MODULE, on_message_acked}),
    % unhook('message.dropped',     {?MODULE, on_message_dropped}).
    % 卸载时需要清除旧的客户端ID
    kafka_stop_client().
   


% del(HookPoint, Action) ->
%    gen_server:cast(?SERVER, {del, HookPoint, Action}).
unhook(HookPoint, Action) -> 
    emqx_hooks:del(HookPoint, Action).

kafka_init(Cnf) ->
    logger:info("Start to init emqx plugin kafka.... ~p~n", [Cnf]),
     % ensure crypto application is load first
     application:ensure_all_started(crypto),
    {ok, _} = application:ensure_all_started(crc32cer),
    {ok, _} = application:ensure_all_started(brod),
    % 获取之前在emqx_plugin_kafka_app.erl 中的kafka 配置
    BootstrapHosts = convert_bootstrap_hosts(maps:get(bootstrap_hosts, Cnf)),
    logger:info("[KAFAK PLUGIN] bootstrap_hosts: ~p", [BootstrapHosts]), %  bootstrap_hosts: [{"10.1.86.200",30092}] mfa: undefined
    KafkaTopic = get_kafka_topic(),
    logger:info("[KAFKA_PLUGIN] kafka_topic: ~p", [KafkaTopic]), % kafka_topic: <<"test">> mfa: undefined
    % 获取配置参数
  
    ReconnectSecs = maps:get(reconnect_cool_down_seconds, Cnf, 10),
    
    % 构建客户端配置
    ClientConfig = [
        {reconnect_cool_down_seconds, ReconnectSecs},
        {query_api_versions, maps:get(query_api_versions, Cnf, true)}
    ],
    logger:info("[KAFKA PLUGIN] client config: ~p", [ClientConfig]),
    case brod:start_client(BootstrapHosts, client1, ClientConfig) of 
      ok -> ok;
      {error, already_present} -> 
        logger:warning("[KAFKA PLUGIN] client1 already present, restarting..."),
        brod:stop_client(client1),
        brod:start_client(BootstrapHosts, client1, []);
      {error, Reason} -> 
        logger:error("[KAFKA PLUGIN] Failed to start kafka client: ~p", [Reason]),
        erlang:error(kafka_client_failed, Reason)
    end,
      %% 修改生产者启动逻辑
    brod:start_producer(client1, KafkaTopic, []), % 这里必须要创建topic 否则会报错
    % 开启客户端 以及生成者订阅
    logger:info("Init emqx plugin kafka successfully ...~n").

kafka_stop_client() ->
    brod:stop_client(client1).


%% ---------------client connect -------------------%%
%  连接成功 解构强制必须有clientid, 发送连接信息到kafka
on_client_connected(ClientInfo = #{clientid := ClientId}, ConnInfo, _Cnf) ->
   {IpAddr, _Port} = maps:get(peername, ConnInfo),
  Now = erlang:system_time(millisecond) div 1000,
   PayLoad = [
    {action, connected},
     {ipaddress, iolist_to_binary(ntoa(IpAddr))},
     {expiry_interval, maps:get(expiry_interval, ConnInfo)},
     {clean_start, maps:get(clean_start, ConnInfo)},
     {sockport, maps:get(sockport, ClientInfo)},
     {connected_at, maps:get(connected_at, ConnInfo)},
     {proto_ver, maps:get(proto_ver, ConnInfo)},
     {proto_name, maps:get(proto_name, ConnInfo)},
     {client_id, ClientId},
     {username, maps:get(username, ClientInfo)},  
     {ts, Now},
     {protocol, maps:get(protocol, ClientInfo)},
     {keepalive, maps:get(keepalive, ConnInfo)}
   ], 
  produce_kafka_payload(ClientId, PayLoad).
  % [KAFKA PLUGIN]Message = {"action":"connected","device_id":"mqttx_19d1691d","username":"undefined","keepalive":60,"ipaddress":"172.17.0.1","proto_name":"MQTT","proto_ver":5,"ts":1749796283450,"online":1}
  %  Client(mqttx_19d1691d) connected,
  %  ClientInfo: , #{clientid => <<"mqttx_19d1691d">>,enable_authn => true,is_bridge => false,is_superuser => false,listener => 'tcp:default'
  %             ,mountpoint => undefined,peerhost => {172,17,0,1},peerport => 46524,protocol => mqtt,
  %           sockport => 1883,username => undefined,zone => default}, , 
  % ConnInfo: ,
  %   #{clean_start => true,clientid => <<"mqttx_19d1691d">>,
  %   conn_mod => emqx_connection,conn_props => #{'Session-Expiry-Interval' => 0},
  %   connected_at => 1749795741542,expiry_interval => 0,keepalive => 60,peercert => nossl,
  %   peername => {{172,17,0,1},46524},proto_name => <<"MQTT">>,proto_ver => 5,receive_maximum => 32,
  %   sockname => {{172,17,0,2},1883},socktype => tcp,username => undefined}, mfa: undefined
  % 这里暂时补充日志， 
  % logger:info("Client(~s) connected, ClientInfo: ~n~p~n, ConnInfo: ~n~p~n", [ClientId, ClientInfo, ConnInfo]).


% 客户端断开连接
on_client_disconnected(ClientInfo = #{clientid := ClientId},ReasonCode, ConnInfo, _Cnf) ->
  {IpAddr, _Port} = maps:get(peername, ConnInfo),
  Now = erlang:system_time(millisecond),
  PayLoad = [
    {action, disconnected},
    {ipaddress, iolist_to_binary(ntoa(IpAddr))},
    {disconnected_at, maps:get(disconnected_at, ConnInfo)},
    {sockport, maps:get(sockport, ClientInfo)},
    {connected_at, maps:get(connected_at, ConnInfo)},
    {proto_ver, maps:get(proto_ver, ConnInfo)},
    {proto_name, maps:get(proto_name, ConnInfo)},
    {client_id, ClientId},
    {username, maps:get(username, ClientInfo)},  
    {ts, Now},
    {protocol, maps:get(protocol, ClientInfo)},
    {reason, ReasonCode}
  ],
  produce_kafka_payload(ClientId, PayLoad).
  % [KAFKA PLUGIN]Message =
  %  {"action":"disconnect","device_id":"mqttx_19d1691d","username":"undefined","reason":"normal","ts":1749796188932,"online":0}
  %
  % Client(mqttx_19d1691d) disconnected due to normal , 
  % ClientInfo: , #{clientid => <<"mqttx_19d1691d">>,enable_authn => true,is_bridge => false,is_superuser => false,listener => 'tcp:default',
  % mountpoint => undefined,peerhost => {172,17,0,1},peerport => 46524,protocol => mqtt,sockport => 1883,username => undefined,zone => default}, , 
  % ConnInfo: , #{clean_start => true,clientid => <<"mqttx_19d1691d">>,conn_mod => emqx_connection,
  % conn_props => #{'Session-Expiry-Interval' => 0},connected_at => 1749795741542,disconn_props => #{},
  % disconnected_at => 1749796177011,expiry_interval => 0,keepalive => 60,peername => {{172,17,0,1},46524},
  % proto_name => <<"MQTT">>,proto_ver => 5,receive_maximum => 32,sockname => {{172,17,0,2},1883},socktype => tcp,username => undefined}, mfa: undefined
  % 日志输出
  % logger:info("Client(~s) disconnected due to ~p , ClientInfo: ~n~p~n, ConnInfo: ~n~p~n", [ClientId, ReasonCode, ClientInfo, ConnInfo]).

%% ---------------message publish -------------------%%
% 该模式匹配会捕获所有的系统消息， 直接返回而不做Kafka转发处理
on_message_publish(Message = #message{topic = <<"$SYS/", _/binary>>}, _Cnf) -> 
    {ok, Message};

on_message_publish(Message, _Cnf) -> 
  % io:format("[KAFKA PLUGIN]before message publish: ~w~n", [Message]),
  %  这里先打印日志 看下
  % mqttx_19d1691d@172.17.0.1:60942 
  % before message publish: 
  % #{extra => #{},flags => #{dup => false,retain => false},from => <<"mqttx_19d1691d">>,
  % headers => #{peerhost => {172,17,0,1},properties => #{},proto_ver => 5,protocol => mqtt,
  % username => undefined},id => <<0,6,55,110,65,52,110,117,247,167,0,0,23,63,0,2>>,
  % payload => <<"{\n  \"msg\": \"hello test plugin\"\n}">>,qos => 0,timestamp => 1749796540149,topic => <<"test">>} mfa: undefined
  %
  %  logger:info("[KAFKA PLUGIN]before message publish: ~p", [emqx_message:to_map(Message)]),
   Topic = emqx_message:topic(Message), % <<"test">>
   {ok, KafkaConfig} = application:get_env(emqx_plugin_kafka, kafka_config),
   case maps:get(mqtt_topics, KafkaConfig, undefined) of 
      undefined -> 
        % 没有找到该配置兼容
        logger:info("mqtt_topics not found in kafka_config, use topic in kafka_config"),
        produce_kafka_msg_original(Message);
      Topics -> 
        logger:info("mqtt_topics found in kafka_config: ~p !", [Topics]),
        % 这里会对mqtt_topics 进行匹配, 如果mqtt发送的topic 和当前配置中的mqtt_topics 一致，返回true
        Matched = lists:any(
          fun(Pattern) -> 
            emqx_topic:match(Topic, Pattern)
          end, Topics
        ),
        % logger:info("mqtt_topics found in config, Topic: ~p, Matched: ~p", [Topic, Matched]),
        case Matched of 
          true -> 
            produce_kafka_msg_original(Message);
          false -> ok
        end
    end,
   {ok, Message}.
produce_kafka_payload(ClientId, Payload) ->
  Topic = get_kafka_topic(),
  % logger:info("[KAFKA PLUGIN]Payload = ~p~n",[Payload]),
  MessageBody = jsx:encode(Payload),
  % io:format("[KAFKA PLUGIN]Message = ~s~n",[MessageBody]),
  % io:format("[KAFKA PLUGIN]Topic = ~s~n",[Topic]),
  % AckCb = fun(Partition, BaseOffset) -> 
  %           io:format(user, "\nProduced to partition ~p at base-offset ~p\n", [Partition, BaseOffset]) 
  %         end,
   AckCb = fun(_Partition, _BaseOffset) -> ok end, 
  brod:produce_cb(client1, Topic, random, ClientId, MessageBody, AckCb).
produce_kafka_msg_original(Message) ->
  {ok, ClientId, PayLoad}  = format_message_payload(Message),
  %  Publish #{extra => #{},
  %        flags => #{dup => false,retain => false},
  %       from => <<"mqttx_19d1691d">>,
  %        headers =>
  %         #{peerhost => {172,17,0,1},
  %              properties => #{},proto_ver => 5,protocol => mqtt,
  %              username => undefined},
  %        id => <<0,6,55,110,123,205,62,179,247,167,0,0,23,224,0,2>>,
  %        payload => <<"{\n  \"msg\": \"hello test plugin\"\n}">>,qos => 0,
  %        timestamp => 1749797523242,topic => <<"test">>}
  produce_kafka_payload(ClientId, PayLoad),
   % 日志输出
  % io:format("Publish ~p~n", [emqx_message:to_map(Message)]),
  {ok, Message}.

format_message_payload(Message) ->
  % Username = emqx_message:get_header(username, Message),
  Topic = Message#message.topic,
  Qos = Message#message.qos,
  % Tail = string:right(binary_to_list(Topic), 4),
  % RawType = string:equal(Tail, <<"_raw">>),
  % io:format("[KAFKA PLUGIN]Tail= ~s , RawType= ~s~n",[Tail,RawType]),
  ClientId = Message#message.from,
  MsgPayload = Message#message.payload,
  io:format("[KAFKA PLUGIN] MsgPayload: ~p~n", [MsgPayload]),
  % 这里打印一下输出的是什么
  MsgPayload64 = list_to_binary(base64:encode_to_string(MsgPayload)),
  io:format("[KAFKA PLUGIN] MsgPayload64: ~p~n", [MsgPayload64]),
  %  io:format("[KAFKA PLUGIN] send message content : ~w~n", [MsgPayload]), 
  %  if
  %   RawType == true ->
  %     MsgPayload64 = list_to_binary(base64:encode_to_string(MsgPayload));
  % ?LOG_INFO("[KAFKA PLUGIN]MsgPayload64 : ~s~n", [MsgPayload64]);
  %   RawType == false ->
  %     MsgPayload64 = MsgPayload
  % end,
  Payload = [
    {action, message},
    {topic, Topic},
    {client_id, ClientId},
    % {username, Username},
    {qos, Qos},
    {payload, MsgPayload64},
    {ts, Message#message.timestamp}],
  {ok, ClientId, Payload}. 

get_kafka_topic() -> 
    % application emqx_plugin_kafka
    % logger:info("all envs: ~p~n", [application:get_all_env(emqx_plugin_kafka)]),
    {ok, Config} =  application:get_env(emqx_plugin_kafka,kafka_config),
    Topic = maps:get(topic, Config),
    Topic.

% <<"bootstrap_hosts">> => [<<"10.1.86.200:30092">>]
convert_bootstrap_hosts(Hosts) when is_list(Hosts) ->
  lists:map(fun convert_bootstrap_host/1, Hosts).

convert_bootstrap_host(Host) when is_binary(Host) ->
  convert_bootstrap_host(binary_to_list(Host));
convert_bootstrap_host(HostStr) ->
  case string:split(HostStr, ":", trailing) of
    [Domain] -> {Domain, 9092};
    [Domain, PortStr] -> {Domain, list_to_integer(PortStr)}
  end.

% 处理ipv6地址
ntoa({0, 0, 0, 0, 0, 16#ffff, AB, CD}) ->
  inet_parse:ntoa({AB bsr 8, AB rem 256, CD bsr 8, CD rem 256});
ntoa(IP) ->
  inet_parse:ntoa(IP).